package my.hadoop.mapreduce.input;

import java.io.IOException;
import java.io.InputStream;

import org.apache.hadoop.io.Text;

public class JsonReader {
	private static final int DEFAULT_BUFFER_SIZE = 64 * 1024;
	private int bufferSize = DEFAULT_BUFFER_SIZE;
	private InputStream in;
	private byte[] buffer;
	// the number of bytes of real data in the buffer
	private int bufferLength = 0;
	// the current position in the buffer
	private int bufferPosn = 0;

	private final byte[] recordDelimiterBytes = "}".getBytes();

	public JsonReader(InputStream in) {
		this(in, DEFAULT_BUFFER_SIZE);
	}

	public JsonReader(InputStream in, int bufferSize) {
		this.in = in;
		this.bufferSize = bufferSize;
		this.buffer = new byte[this.bufferSize];
	}

	public int readJsonStr(Text str, int maxStrLength, int maxBytesToConsume)
			throws IOException {
		str.clear();
		int txtLength = 0; // tracks str.getLength(), as an optimization
		long bytesConsumed = 0;
		int delPosn = 0;
		do {
			int startPosn = bufferPosn; // starting from where we left off the
										// last
			// time
			if (bufferPosn >= bufferLength) {
				startPosn = bufferPosn = 0;
				bufferLength = in.read(buffer);
				if (bufferLength <= 0)
					break; // EOF
			}
			for (; bufferPosn < bufferLength; ++bufferPosn) {
				if (buffer[bufferPosn] == recordDelimiterBytes[delPosn]) {
					delPosn++;
					if (delPosn >= recordDelimiterBytes.length) {
						bufferPosn++;
						break;
					}
				} else {
					delPosn = 0;
				}
			}
			int readLength = bufferPosn - startPosn;
			bytesConsumed += readLength;
			// TODO
			int appendLength = readLength - delPosn;
			if (appendLength > maxStrLength - txtLength) {
				appendLength = maxStrLength - txtLength;
			}
			if (appendLength > 0) {
				str.append(buffer, startPosn, appendLength);
				str.set(str.toString() + "}");
				txtLength += appendLength;
			}

			//Skip json object separator
			bufferPosn++;
		} while (delPosn < recordDelimiterBytes.length
				&& bytesConsumed < maxBytesToConsume);
		if (bytesConsumed > (long) Integer.MAX_VALUE)
			throw new IOException("Too many bytes before delimiter: "
					+ bytesConsumed);
		return (int) bytesConsumed;
	}

	public void close() throws IOException {
		in.close();
	}
}
